// mqtt project main.go
package main

import (
	"encoding/json"
	"flag"
//	"fmt"
	"os"
	"os/signal"
//	"strconv"
//	"time"
	"github.com/astaxie/beego/logs"
	MQTT "github.com/eclipse/paho.mqtt.golang"
)

var Log *logs.BeeLogger

func main() {
	topic := flag.String("topic", "monitor/41010501302", "The topic name to/from which to publish/subscribe")
	broker := flag.String("broker", "tcp://121.42.182.8:1883", "The broker URI. ex: tcp://10.10.1.1:1883")
	password := flag.String("password", "", "The password (optional)")
	user := flag.String("user", "", "The User (optional)")
	id := flag.String("id", "datareceivertmp", "The ClientID (optional)")
	cleansess := flag.Bool("clean", true, "Set Clean Session (default false)")
	qos := flag.Int("qos", 0, "The Quality of Service 0,1,2 (default 0)")
	store := flag.String("store", ":memory:", "The Store Directory (default use memory store)")
	//mongo = flag.String("mongo", "root:123456@localhost:27017", "mongo connstr")
	flag.Parse()

	//InitMongo(*mongo)

	Log = logs.NewLogger(10000)
	Log.SetLogger("console", "")
	Log.EnableFuncCallDepth(true)

	//os.MkdirAll("c:/logs/dataReceiver", os.ModeDir)
	Log.SetLogger("file", `{"filename":"`+"c:/logs/dataReceiver/dataReceiver.log"+`"}`)
	Log.SetLogger("smtp", `{"level":2,"subject":"恒大","fromAddress":"songkemail@163.com","username":"songkemail@163.com","password":"songke198916","host":"smtp.163.com:25","sendTos":["553403792@qq.com"]}`)

	//cd := make(chan map[string]float64, 1)
	//go consumer(cd)

	opts := MQTT.NewClientOptions()
	opts.SetAutoReconnect(true)
	opts.AddBroker(*broker)
	opts.SetClientID(*id)
	opts.SetUsername(*user)
	opts.SetPassword(*password)
	opts.SetCleanSession(*cleansess)
	opts.OnConnect = func(client MQTT.Client) {
		if token := client.Subscribe(*topic, byte(*qos), func(c MQTT.Client, m MQTT.Message) {
			var data map[string]float64
			if err := json.Unmarshal(m.Payload(), &data); err != nil {
				Log.Error("Unmarshal err:%v", err)
				return
			}
			//cd <- data
			Log.Debug("cd <- data topic:%v, msg:%v", m.Topic(), string(m.Payload()))
		}); token.Wait() && token.Error() != nil {
			Log.Error("Subscribe token.Error:%v", token.Error())
			os.Exit(1)
		}
	}
	if *store != ":memory:" {
		opts.SetStore(MQTT.NewFileStore(*store))
	}

	// Set up channel on which to send signal notifications.
	sigc := make(chan os.Signal, 1)
	signal.Notify(sigc, os.Interrupt, os.Kill)

	client := MQTT.NewClient(opts)
	if token := client.Connect(); token.Wait() && token.Error() != nil {
		panic(token.Error())
	}
	defer client.Disconnect(250)
	<-sigc
}
